package com.atguigu.realtime.dws

import com.atguigu.realtime.BaseAppV3
import com.atguigu.realtime.bean.{OrderDetail, OrderInfo, OrderWide}
import com.atguigu.realtime.util.MyRedisUtil
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.kafka010.OffsetRange
import org.json4s.jackson.{JsonMethods, Serialization}
import redis.clients.jedis.Jedis

import scala.collection.JavaConverters._
import scala.collection.mutable.ListBuffer

/**
 * Author atguigu
 * Date 2020/11/20 9:09
 */
object DwsOrderWideApp_2 extends BaseAppV3 {
    override val master: String = "local[2]"
    override val appName: String = "DwsOrderWideApp_2"
    override val groupId: String = "DwsOrderWideApp_2"
    override val topics: Seq[String] = Seq("dwd_order_info", "dwd_order_detail")
    override val bachTime: Int = 5
    
    private val orderInfoPre = "order_info"
    private val orderDetailPre = "order_detail"
    
    /**
     * 缓存OrderDetail
     *
     * @param client
     * @param orderDetail
     * @return
     */
    def cacheOrderDetail(client: Jedis, orderDetail: OrderDetail) = {
        implicit val f = org.json4s.DefaultFormats
        val key = s"${orderDetailPre}:${orderDetail.order_id}"
        val field = orderDetail.id.toString
        val value = Serialization.write(orderDetail)
        client.hset(key, field, value)
    }
    
    /**
     * 缓村OrderInfo信息
     *
     * @param client
     * @param orderInfo
     * @return
     */
    def cacheOrderInfo(client: Jedis, orderInfo: OrderInfo) = {
        implicit val f = org.json4s.DefaultFormats
        val key = s"${orderInfoPre}:${orderInfo.id}"
        val value = Serialization.write(orderInfo)
        client.setex(key, 10 * 60, value)
    }
    
    override def run(ssc: StreamingContext,
                     sourceStreams: Map[String, DStream[ConsumerRecord[String, String]]],
                     offsetRanges: Map[String, ListBuffer[OffsetRange]]): Unit = {
        val orderInfoStream = sourceStreams("dwd_order_info")
            .map(record => {
                implicit val f = org.json4s.DefaultFormats + toLong + toDouble
                val orderInfo: OrderInfo = JsonMethods.parse(record.value()).extract[OrderInfo]
                (orderInfo.id, orderInfo)
            })
        
        val orderDetailStream = sourceStreams("dwd_order_detail")
            .map(record => {
                implicit val f = org.json4s.DefaultFormats + toLong + toDouble
                val orderDetail: OrderDetail = JsonMethods.parse(record.value()).extract[OrderDetail]
                (orderDetail.order_id, orderDetail)
            })
        val orderWideDetail = orderInfoStream
            .fullOuterJoin(orderDetailStream)
            .mapPartitions(it => {
                implicit val f = org.json4s.DefaultFormats
                val client: Jedis = MyRedisUtil.getClient
                
                val result = it.flatMap {
                    case (orderId, (Some(orderInfo), Some(orderDetail))) =>
                        println("some some")
                        
                        // 1. 把orderInfo存入缓存
                        cacheOrderInfo(client, orderInfo)
                        // 2. 和同批次的join
                        val orderWide = new OrderWide(orderInfo, orderDetail)
                        
                        // 3. 读对方缓存
                        val key = s"${orderDetailPre}:${orderId}"
                        val r = client.hgetAll(key).asScala
                            .map {
                                case (orderDetailId, orderDetailStr) =>
                                    val orderDetail = JsonMethods.parse(orderDetailStr).extract[OrderDetail]
                                    new OrderWide(orderInfo, orderDetail)
                            }
                            .toList
                        client.del(key)
                        //  r :+ orderWide
                        //  orderWide +: r
                        orderWide :: r
                    case (orderId, (Some(orderInfo), None)) =>
                        println("some none")
                        // 1. 把orderInfo存入缓存
                        cacheOrderInfo(client, orderInfo)
                        // 2. 去对方缓存(OrderDetail的缓存)
                        val key = s"${orderDetailPre}:${orderId}"
                        val r = client.hgetAll(key).asScala
                            .map {
                                case (orderDetailId, orderDetailStr) =>
                                    val orderDetail = JsonMethods.parse(orderDetailStr).extract[OrderDetail]
                                    new OrderWide(orderInfo, orderDetail)
                            }
                            .toList
                        client.del(key)
                        r
                    case (orderId, (None, Some(orderDetail))) =>
                        println("none some")
                        // 1. 去order_info缓存查找
                        val orderInfoStr: String = client.get(s"${orderInfoPre}:${orderId}")
                        if (orderInfoStr != null) { // 2. 如果找到join
                            val orderInfo: OrderInfo = JsonMethods.parse(orderInfoStr).extract[OrderInfo]
                            new OrderWide(orderInfo, orderDetail) :: Nil
                        } else { // 3. 如果没找到, 把自己存入缓存
                            cacheOrderDetail(client, orderDetail)
                            Nil
                        }
                }
                
                client.close()
                result
            })
        orderWideDetail.print(1000)
        
        /*.foreachRDD(rdd => {
            println("foreachRDD...")
            rdd.collect().foreach(println)
            
            OffsetManager.saveOffsets(offsetRanges.values.reduce(_ ++ _), groupId, topics)
        })
    */
    }
}

/*
order_info缓存类型:
    容易存, 容易读
    key                             value
    "order_info:"+order_id          OrderInfo的json字符串格式  {"": "", "": ""}
 
 
order_detail缓存类型:
    容易存, 容易读
    
    key                                                  value
    //"order_detail:"+order_id+order_detail_id             OrderDetail的json字符串格式  {"": "", "": ""}
    
    "order_detail:"+order_id                             value(hash)
                                                         field                  value
                                                         order_detail_id        OrderDetail的json字符串格式  {"": "", "": ""}

 */